package com.sjk.flink.functions;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.internal.JsonContext;
import com.sjk.flink.correct.IfCorrectRule;
import com.sjk.flink.correct.rules.UnifyDateTimeRuler;
import com.sjk.flink.exeception.ErrorCode;
import com.sjk.flink.json.KeyConverter;
import com.sjk.flink.pojo.*;
import com.sjk.flink.util.*;
import networknt.schema.ValidationMessage;
import com.sjk.flink.exeception.BaseException;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;
//import com.fasterxml.jackson.core.exc.StreamConstraintsException;

/**
 * @author SAMA
 */
public class JsonValidationProcessFunction extends ProcessFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>> {

    private static final Logger log = LoggerFactory.getLogger(JsonValidationProcessFunction.class);

    private final OutputTag<OutPutPojo> validJsonTag;
    private final OutputTag<OutPutPojo> invalidJsonTag;

    /** 共通错误的对应的topic **/
    private final OutputTag<OutPutPojo> commonInvalidJsonTag;

    //private final OutputTag<Performance> performanceJsonTag;

    //private final OutputTag<OutPutPojo> expireJsonTag;

    private transient ObjectMapper objectMapper;

    private transient ParameterTool parameterTool;

    @Override
    public void open(Configuration parameters) {
        objectMapper = ObjectMapperUtil.getObjectMapper();
        parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
    }

    private final Map<String,String> SCHEMA_CONFIGS;

    private final Map<String,String> JZZL_CONFIGS;

    private boolean zlfxFlg;

    /**
     *
     * @param zlfxFlg 质量分析标志
     * @param SCHEMA_CONFIGS json schema 配置
     * @param JZZL_CONFIGS   质量分析配置
     * @param validJsonTag
     * @param invalidJsonTag
     * @param commonInvalidJsonTag
     */
    public JsonValidationProcessFunction(boolean zlfxFlg,Map<String,String> SCHEMA_CONFIGS, Map<String,String> JZZL_CONFIGS, OutputTag<OutPutPojo> validJsonTag, OutputTag<OutPutPojo> invalidJsonTag, OutputTag<OutPutPojo> commonInvalidJsonTag) {
        this.validJsonTag = validJsonTag;
        this.invalidJsonTag = invalidJsonTag;
        this.commonInvalidJsonTag = commonInvalidJsonTag;
        //this.performanceJsonTag = performanceJsonTag;
        this.SCHEMA_CONFIGS = SCHEMA_CONFIGS;
        this.JZZL_CONFIGS = JZZL_CONFIGS;
        this.zlfxFlg= zlfxFlg;
        //this.expireJsonTag = expireJsonTag;
    }


    private String generateInvalidJsonString(String data, String reason) throws JsonProcessingException {
        ObjectNode rootNode = objectMapper.createObjectNode();
        rootNode.put("data", "data");
        rootNode.put("extract_time", JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));
        rootNode.put("checkResult", reason);
        return objectMapper.writeValueAsString(rootNode);
    }

    @Override
    public void processElement(ConsumerRecord<String, String> stringStringConsumerRecord, ProcessFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>>.Context ctx, Collector<ConsumerRecord<String, String>> collector) throws Exception {
        String value = stringStringConsumerRecord.value();
        String topic = stringStringConsumerRecord.topic();
        //log.info("来源topic:                  " + topic);
        System.out.println("来源topic:                  " + topic);
        int  clayer_partition = stringStringConsumerRecord.partition();
        long  clayer_offset = stringStringConsumerRecord.offset();
        Long  clayer_ts = stringStringConsumerRecord.timestamp();
        /**
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        String clayer_timestamp_str = sdf.format(new Date(clayer_timestamp));
         **/
        String zxdm = parameterTool.get("zxdm");
        System.out.println("专项代码                 " + zxdm);
        //获取表名 去除ODS_之前的字符
        String tableName = topic.substring(topic.indexOf("ODS_")+4).trim();

        String topicsscsdm = "";
        String start_time = JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss");

        long startTime = System.currentTimeMillis();
        OutPutPojo outPutPojo = new OutPutPojo();
        outPutPojo.setTableName(tableName);

        //城市分区
        int city_partition_id = 0;
        //城市偏移量
        long city_offset_id = 0;
        //时间戳
        String city_ts = "";

        try {
            JsonNode readTree = objectMapper.readTree(value);
            JsonNode sscsdmJsonNode = readTree.get("sscsdm");
            //城市代码
            topicsscsdm = sscsdmJsonNode.asText();
            //抽取时间
            JsonNode extract_timeJsonNode = readTree.get("extract_time");
            String extract_time = extract_timeJsonNode.asText();
            //城市分区
            JsonNode city_partition_idJsonNode = readTree.get("city_partition_id");
            if(city_partition_idJsonNode!=null) {
                city_partition_id = new Integer(city_partition_idJsonNode.asText());
            }
            //城市偏移量
            JsonNode city_offset_idJsonNode = readTree.get("city_offset_id");
            if(city_offset_idJsonNode!=null) {
                city_offset_id = new Long(city_offset_idJsonNode.asText());
            }
            //时间戳
            JsonNode city_tsJsonNode = readTree.get("city_ts");
            if(city_tsJsonNode!=null) {
                city_ts = city_tsJsonNode.asText();
            }
            //过期校验标志
            //boolean verify = parameterTool.getBoolean(topic+".verify.flag",false);

            JsonNode recordJsonNode = readTree.get("record");
            String srcData1 = recordJsonNode.asText();
            if(MessyCodeCheck.isMessyCode(srcData1)){
                //乱码
                throw new BaseException(ErrorCode.CODE_E00004);
            }
            //获得record数据
            JsonNode jsonNode = objectMapper.readTree(srcData1);
            if(!zlfxFlg) {
                //不做数据校准及质量分析直接到标准库
                boolean isObj = jsonNode.isObject();
                boolean isArray = jsonNode.isArray();
                if (isArray) {
                    for (JsonNode node1 : jsonNode) {
                        //转为小写
                        JsonNode lowerCaseNode = KeyConverter.convertKeysToLowercase(node1);
                        String arrayJsonStr = objectMapper.writeValueAsString(lowerCaseNode);
                        outPutPojo.setJson(packageJson(topicsscsdm, extract_time, arrayJsonStr, city_partition_id, city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
                        //验证成功
                        ctx.output(validJsonTag, outPutPojo);
                    }
                } else if (isObj) {
                    //字段名转小写
                    JsonNode lowerCaseNode = KeyConverter.convertKeysToLowercase(jsonNode);
                    String arrayJsonStr = objectMapper.writeValueAsString(lowerCaseNode);
                    outPutPojo.setJson(packageJson(topicsscsdm, extract_time, arrayJsonStr,city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
                    //验证成功
                    ctx.output(validJsonTag, outPutPojo);
                } else {
                    System.out.println("不是有效的json");
                    log.info("不是有效的json");
                    outPutPojo.setJson(analysisExecption(topic, topicsscsdm, value, "000000","不是有效的json数据",city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
                    //不是有效的json
                    //不是有效的 JSON，发送到无效数据的侧输出
                    ctx.output(commonInvalidJsonTag, outPutPojo);
                }

            }else {
                System.out.println("... ... 数据校准开始 ... ...");
                //主题检查
                JsonChecker odsJsonChecker = JsonSchemaUtil.getPatternIdByKey(SCHEMA_CONFIGS, "ODS", value);
                if (!odsJsonChecker.isSuccess()) {
                    outPutPojo.setJson(objectMapper.writeValueAsString(odsJsonChecker));
                    //检查失败
                    ctx.output(invalidJsonTag, outPutPojo);
                } else {
                    //数据校准
                    //判断数据是否为有效JSON
                    boolean isObj = jsonNode.isObject();
                    boolean isArray = jsonNode.isArray();
                    if (isArray) {

                        //拆分数组
                        int i = 0;
                        for (JsonNode node1 : jsonNode) {
                            //转小写
                            JsonNode lowerCaseNode = KeyConverter.convertKeysToLowercase(node1);
                            processRecord(ctx,lowerCaseNode,tableName,topic,outPutPojo,zxdm,topicsscsdm,extract_time,city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts);
                            i++;
                        }
                        System.out.println("数组长度: >>>>>>>>>>>>>>>>>  "+i);
                    } else if (isObj) {
                        //转小写
                        JsonNode lowerCaseNode = KeyConverter.convertKeysToLowercase(jsonNode);
                        processRecord(ctx,lowerCaseNode,tableName,topic,outPutPojo,zxdm,topicsscsdm,extract_time,city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts);
                    } else {
                        //System.out.println("不是有效的json");
                        //log.info("不是有效的json");
                        outPutPojo.setJson(analysisExecption(topic, topicsscsdm, value, "000000","不是有效的json数据",city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
                        //不是有效的json
                        //不是有效的 JSON，发送到无效数据的侧输出
                        ctx.output(commonInvalidJsonTag, outPutPojo);
                    }
                }
            }
            /**
            Performance performance = new Performance();
            performance.setExecute_state("正常");
            performance.setStart_time(start_time);
            performance.setSrcdata(value);
            performance.setTopic(tableName);
            long executTime = System.currentTimeMillis()-startTime;
            performance.setExecute_time(executTime);
            ctx.output(performanceJsonTag, performance);
             **/
        } catch (Exception e) {
            e.getStackTrace();
            String errorCode = e.getClass().getName();
            if(e instanceof BaseException){
                errorCode = ((BaseException) e).getCode();
            }
            //城市代码可能为空
            outPutPojo.setJson(analysisExecption(topic,topicsscsdm,value,errorCode,"解析错误:"+ ExecptionUtil.getStackTraceMsg(e),city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
            //解析 JSON 出错，发送到无效数据的侧输出
            ctx.output(commonInvalidJsonTag, outPutPojo);
            //执行失败
            /**
            Performance performance = new Performance();
            performance.setExecute_state("正常");
            performance.setStart_time(start_time);
            performance.setSrcdata(value);
            performance.setTopic(tableName);
            long executTime = System.currentTimeMillis()-startTime;
            performance.setExecute_time(executTime);
            ctx.output(performanceJsonTag, performance);
             **/
        }
    }

    /**
     * 拆分记录执行
     *
     * @param ctx
     * @param lowerCaseNode
     * @param tableName
     * @param srcTopic
     * @param outPutPojo
     * @param zxdm
     * @param topicsscsdm
     * @param extract_time
     * @throws Exception
     */
    private void processRecord(ProcessFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>>.Context ctx,JsonNode lowerCaseNode,String tableName,String srcTopic,OutPutPojo outPutPojo,String zxdm,String topicsscsdm,String extract_time,int city_partition_id,long city_offset_id,String city_ts,int clayer_partition,long clayer_offset,long clayer_ts)throws Exception{
        String arrayJsonStr = objectMapper.writeValueAsString(lowerCaseNode);
        //2、校准值替换
        JzPojo jzpojo = JzValueUtils.jsonhander(JZZL_CONFIGS, tableName, arrayJsonStr);
        String jzJsonStr = jzpojo.getJson();
        System.out.println("校准后的数据                         " + jzJsonStr);
        //校准标志 0 未校准  1 校准成功  2 校准失败
        String jzbz = "0";
        //校对信息推送
        if(!jzpojo.isSkip()){
            if(jzpojo.isSuccess()){
                //校准成功
                jzbz = "1";
            }else{
                //校准失败
                jzbz = "2";
                List<JzField> fields = jzpojo.getFields();
                if (fields.size() > 0) {
                    for (JzField jzField : fields) {
                        //把错误信息发送到ECT
                        outPutPojo.setJson(analysisExecption(srcTopic, topicsscsdm, jzpojo.getJson(), jzField.getErrorCode(), jzField.getErrorMesasge(),city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
                        //解析 JSON 出错，发送到无效数据的侧输出
                        ctx.output(commonInvalidJsonTag, outPutPojo);
                    }
                } else {
                    //把错误信息发送到ECT
                    outPutPojo.setJson(analysisExecption(srcTopic, topicsscsdm, jzpojo.getJson(), jzpojo.getErrorCode(), jzpojo.getErrorMesasge(),city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
                    //解析 JSON 出错，发送到无效数据的侧输出
                    ctx.output(commonInvalidJsonTag, outPutPojo);
                }
            }
        }

        JsonNode jcsjzjJsonNode = lowerCaseNode.get("bsm");
        String jcsjzj = null;
        if(jcsjzjJsonNode!=null){
            jcsjzj = jcsjzjJsonNode.asText();
        }

        //检查是否
        JsonChecker srcDataJsonChecker = JsonSchemaUtil.getPatternIdByKey(SCHEMA_CONFIGS, srcTopic, jzJsonStr);
        if (!srcDataJsonChecker.isSuccess()) {
            //获得校对质检文件
            JsonContext jzzlDocument = (JsonContext) JsonPath.parse(JZZL_CONFIGS.get(tableName));
            //需要校准的文件
            JsonContext srcDocument = (JsonContext) JsonPath.parse(jzJsonStr);

            Set<ValidationMessage> validates = srcDataJsonChecker.getValidate();
            //获得所有的字段分级
            boolean stdout = true;
            List<String> zdfjs =getZdfjs(jzzlDocument,validates);
            if(zdfjs.isEmpty()||zdfjs.contains("A")){
                //写问题
                stdout = false;
            }
            //字段错误处理
            Iterator<ValidationMessage> iterator = validates.iterator();
            while (iterator.hasNext()) {
                ValidationMessage next = iterator.next();
                outPutPojo.setJson(analysis(jzbz,zxdm, jcsjzj, topicsscsdm, extract_time, srcTopic, srcDataJsonChecker.getExtract_time(), srcDocument, jzzlDocument, next,city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
                //验证失败
                ctx.output(invalidJsonTag, outPutPojo);
            }
            //全B类数据同时输出到标准库
            if(stdout){
                outPutPojo.setJson(packageJson(topicsscsdm, extract_time, srcDataJsonChecker.getData(),city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
                //验证成功
                ctx.output(validJsonTag, outPutPojo);
            }
        } else {
            //System.out.println("检查成功     " + JSON.toJSONString(srcDataJsonChecker));
            //log.info("检查成功     " + JSON.toJSONString(srcDataJsonChecker));
            outPutPojo.setJson(packageJson(topicsscsdm, extract_time, srcDataJsonChecker.getData(),city_partition_id,city_offset_id,city_ts,clayer_partition,clayer_offset,clayer_ts));
            //验证成功
            ctx.output(validJsonTag, outPutPojo);
        }
    }

    /**
     * 封装成功返回的JSON
     *
     * @param sscddm
     * @param extract_time
     */
    private String packageJson(String sscddm,String extract_time,String scrData,int city_partition_id,long city_offset_id,String city_ts,int clayer_partition,long clayer_offset,Long clayer_ts) throws JsonProcessingException {
        ObjectNode readTree = (ObjectNode)objectMapper.readTree(scrData);
        readTree.put("sscsdm",sscddm);
        readTree.put("extract_time",extract_time);
        readTree.put("operate_time",JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));
        readTree.put("city_partition_id",city_partition_id);
        readTree.put("city_offset_id",city_offset_id);
        readTree.put("city_ts",city_ts);
        readTree.put("clayer_partition",clayer_partition);
        readTree.put("clayer_offset",clayer_offset);
        if(clayer_ts==null){
            readTree.put("clayer_ts", "");
        }else{
            readTree.put("clayer_ts", formatDate(clayer_ts));
        }
        return objectMapper.writeValueAsString(readTree);
    }

    private List<String> getZdfjs(JsonContext jzzlDocument,Set<ValidationMessage> validates){
        List<String> zdfjs =new ArrayList<String>();
        Iterator<ValidationMessage> iterator = validates.iterator();
        while (iterator.hasNext()) {
            ValidationMessage next = iterator.next();
            //json path
            String path = next.getPath();
            String message = new String(next.getMessage().getBytes(StandardCharsets.ISO_8859_1), StandardCharsets.UTF_8);
            if(path.equals("$")){
                //分割
                String[] msgs = message.split(":");
                path = msgs[0];
            }
            String zdfiPath = path+".zdfj";
            try {
                String zdfj = jzzlDocument.read(zdfiPath, String.class);
                zdfjs.add(zdfj);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return zdfjs;
    }

    /**
     * 质量分析
     *
     * @param jcsjzj
     * @param topicsscsdm
     * @param extractTime
     * @param topic
     * @param zjsj
     * @param srcDocument
     * @param jzzlDocument
     * @param validationMessage
     * @return
     */
    private String analysis(String jzbz,String zxdm,String jcsjzj,String topicsscsdm,String extractTime, String topic,String zjsj,JsonContext srcDocument,JsonContext jzzlDocument, ValidationMessage validationMessage,int city_partition_id,long city_offset_id,String city_ts,int clayer_partition,long clayer_offset,Long clayer_ts) throws JsonProcessingException {
        //json path
        String path = validationMessage.getPath();
        String code = validationMessage.getCode();
        String message = validationMessage.getMessage();
        //String message = new String(validationMessage.getMessage().getBytes(StandardCharsets.ISO_8859_1), StandardCharsets.UTF_8);
        if(path.equals("$")){
            //分割
            String[] msgs = message.split(":");
            //TODO 检查
            path = msgs[0];
            message = new String(validationMessage.getMessage().getBytes(StandardCharsets.ISO_8859_1), StandardCharsets.UTF_8);
        }
        //取值
        //JsonContext document = (JsonContext) JsonPath.parse(srcJson);

        //分析结果
        QSTTY qstty = new QSTTY();
        //校准标志
        qstty.setJzbz(jzbz);
        /** 检测数据主键(标识码)" **/
        qstty.setJcsjzj(jcsjzj);

        /** topic名称 **/
        qstty.setTopicmc(topic);

        /** 专项代码 **/
        qstty.setZxdm(zxdm);

        /** 城市 **/
        qstty.setCs(topicsscsdm);
        //获取原始值
        try {
            Object value = srcDocument.read(path);
            /** 问题字段原始值 **/
            qstty.setWtzdysz(String.valueOf(value));
        }catch (Exception ex){
            qstty.setWtzdysz(null);
        }

        /** 质检规则 **/
        //根据code获得质检规则
        /**
        String zjgzmsg = StreamProcess.properties.getProperty(code);
        String zjgz ="";
        String zjgzfl ="";
        if(zjgzmsg!=null){
            String[] msgs =zjgzmsg.split("|");
            if(msgs.length==2){
                zjgz = msgs[0];
                zjgzfl= msgs[1];
            }else{
                zjgz = zjgzmsg;
            }
        }
         **/
        String zjgz =validationMessage.getType();
        String zjgzfl =message;
        qstty.setZjgz(zjgz);

        /** 字段分级 **/
        String zdfiPath = path+".zdfj";
        try {
            qstty.setZdfj(jzzlDocument.read(zdfiPath, String.class));
        }catch (Exception e){
            e.printStackTrace();
        }

        /** 质检规则分类 **/
        qstty.setZjgzfl(zjgzfl);

        /** 质检时间 **/
        qstty.setZjsj(zjsj);

        /** 操作时间 **/
        qstty.setOperate_time(JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));

        /** 抽取时间 **/
        qstty.setExtract_time(extractTime);

        /** 问题字段名称 **/
        qstty.setWtzdmc(path.replace("$.",""));

        /** 错误code **/
        qstty.setErrorCode(code);

        /** 错误消息 **/
        qstty.setErrorMessage(message);
        qstty.setCity_partition_id(city_partition_id);
        qstty.setCity_offset_id(city_offset_id);

        qstty.setCity_ts(city_ts);

        qstty.setClayer_partition(clayer_partition);
        qstty.setClayer_offset(clayer_offset);
        if(clayer_ts==null){
            qstty.setClayer_ts("");
        }else{
            qstty.setClayer_ts(formatDate(clayer_ts));
        }

        String json = objectMapper.writeValueAsString(qstty);
        System.out.println("####################校验错误输出#####################");
        System.out.println(json);
        System.out.println("#########################################");
        return json;
    }


    private String formatDate(long s){
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        // 设置时区为东八区（北京）
        TimeZone timeZone = TimeZone.getTimeZone("GMT+8");
        sdf.setTimeZone(timeZone);

        // 获取当前时间
        Date date = new Date(s);

        // 将日期按照指定的格式转换成字符串
        return sdf.format(date);
    }

    /**
     * 解析及系统错误分析
     *
     * @param topic
     * @param cs
     * @param srcData
     * @return
     */
    private String analysisExecption(String topic,String cs,String srcData,String code,String message,int city_partition_id,long city_offset_id,String city_ts,int clayer_partition,long clayer_offset,Long clayer_ts) throws JsonProcessingException {
        ECTTY ectty  =new ECTTY();
        /** topic名称 **/
        ectty.setTopicmc(topic);

        /** 城市 **/
        ectty.setCs(cs);

        /** 原始数据 **/
        ectty.setScrData(srcData);

        /** 错误代码 **/
        ectty.setErrorCode(code);
        /** 错误信息 **/
        ectty.setErrorMessage(message);
        ectty.setCity_partition_id(city_partition_id);
        ectty.setCity_offset_id(city_offset_id);
        ectty.setCity_ts(city_ts);

        ectty.setClayer_partition(clayer_partition);
        ectty.setClayer_offset(clayer_offset);
        if(clayer_ts==null){
            ectty.setClayer_ts("");
        }else{
            ectty.setClayer_ts(formatDate(clayer_ts));
        }

        ectty.setOperate_time(JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));
        String json = objectMapper.writeValueAsString(ectty);
        return json;
    }
}
